package hudi;

import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/10/22
 */
public class ReadHudi {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000L);
        env.setStateBackend(new FsStateBackend("hdfs://172.17.1.103:8020/hudi/test/xyc_test_json_04/state"));
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(env, bsSettings);

        bsTableEnv.executeSql(
                "create table hudi_table (id int, gender string, name string, ts TIMESTAMP(3),PRIMARY KEY (`id`) NOT ENFORCED) partitioned by (gender)" +
                        "with ('connector'='hudi', 'path'='hdfs://172.17.1.103:8020/hudi/test/xyc_test_json_04'," +
                        "'table.type'='MERGE_ON_READ')");

        TableResult tableResult = bsTableEnv.executeSql("select * from hudi_table");
        tableResult.print();
    }
}
